package com.learn.netty.reactor;

import com.learn.netty.bio.multithread.PrintUtils;
import io.netty.util.concurrent.DefaultThreadFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 响应器
 */
public class Reactor implements Runnable {
    // 选择器
    final Selector selector;
    // 服务端socketChannel
    final ServerSocketChannel serverSocketChannel;
    final ExecutorService executor = Executors.newCachedThreadPool(new DefaultThreadFactory("handler-thread"));

    public Reactor(int port) throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        InetSocketAddress address = new InetSocketAddress(port);
        serverSocketChannel.socket().bind(address);
        PrintUtils.printLog(String.format("serverSocket bind address, ip:%s port:%d", address.getHostName(), address.getPort()));
        serverSocketChannel.configureBlocking(false);
        SelectionKey selectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        PrintUtils.printLog("selector register serverSocketChannel -> OP_ACCEPT");
        selectionKey.attach(new Acceptor());
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                selector.select();
                PrintUtils.printLog("selector select");
                Set<SelectionKey> selected = selector.selectedKeys();
                selected.forEach(this::dispatch);
                selected.clear();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void dispatch(SelectionKey selectionKey) {
        Runnable r = (Runnable) selectionKey.attachment();
        if (r != null) {
            r.run();
        }
    }

    class Acceptor implements Runnable {

        @Override
        public void run() {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                if (socketChannel != null) {
                    PrintUtils.printLog(String.format("serverSocketChannel accept, remoteAddress:%s", socketChannel.getRemoteAddress()));
                    executor.submit(new Handler(selector, socketChannel));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
